package com.shujia.tour

import com.ctyun.daas.common.{CalculateLength, Geography}
import com.ctyun.daas.common.poly.Polygon
import org.apache.spark.sql.expressions.{UserDefinedFunction, Window}
import org.apache.spark.sql.types.DataType
import org.apache.spark.sql.{DataFrame, Dataset, Row, SaveMode, SparkSession}

object DALTourScenicMskDay {
  def main(args: Array[String]): Unit = {

    //获取时间参数
    val day_id: String = args.head

    val month_id: String = day_id.substring(0, 6)

    println(s"时间参数：day_id = $day_id,month_id=$month_id")

    //1、创建环境
    val spark: SparkSession = SparkSession
      .builder()
      .appName("DALTourScenicMskDay")
      //开启hive元数据支持
      .enableHiveSupport()
      .getOrCreate()
    import spark.implicits._
    import org.apache.spark.sql.functions._

    //2、读取停留表表
    val stayPoint: DataFrame = spark
      .table("dwi.dwi_evt_oidd_staypoint_msk_d")
      .where($"day_id" === day_id)

    //4、景区边界配置表
    val scenicBoundary: DataFrame = spark
      .table("dim.dim_map_scenic_boundary_m")
      .where($"month_id" === month_id)

    //网格配置表
    val geotagGrid: DataFrame = spark
      .table("dim.dim_geotag_grid")

    val usertag: DataFrame = spark
      .table("dim.dim_usertag_msk_d")
      .where($"day_id" === day_id)

    val adminCode: DataFrame = spark
      .table("dim.dim_map_admin_code_m")
      .where($"month_id" === month_id)

    //关联获取常住地市
    val resiCity: DataFrame = usertag
      .join(adminCode, $"resi_county_id" === $"county_id")
      .join(geotagGrid, $"resi_grid_id" === $"grid_id")
      .select($"mdn", $"city_id" as "source_city_id", $"center_longi".cast("double") as "resi_lon", $"center_lati".cast("double") as "resi_lat")


    //定义自定义函数，判断点是否在边界内
    val polygon_contains: UserDefinedFunction = udf((lon: String, lat: String, boundary: String) => {
      val polygon = new Polygon(boundary)
      polygon.contains(lon.toDouble, lat.toDouble)
    })

    //定义计算距离的函数
    val calculate_length = udf((long1: Double, lat1: Double, long2: Double, lat2: Double) => {
      Geography.calculateLength(long1, lat1, long2, lat2)
    })

    //计算景区内的网格
    val scenicGrid: DataFrame = scenicBoundary.hint("broadcast")
      .join(geotagGrid)
      .where(polygon_contains($"center_longi", $"center_lati", $"boundary"))
      .select($"scenic_id", $"scenic_name", $"grid_id")

    //关联停留表判断停留点是否出现在景区
    val resultDF: Dataset[Row] = stayPoint
      //关联常住地
      .join(resiCity, "mdn")
      //计算每隔停留点到常住地的距离
      .withColumn("distance", round(calculate_length($"resi_lon", $"resi_lat", $"lon", $"lat") / 1000, 2))
      //本次出游到目前为止出游最远距离（公里）
      .withColumn("d_max_distance", max($"distance") over (Window.partitionBy($"mdn")))
      //计算出发时间
      .withColumn("departure_time", min($"e_time") over Window.partitionBy($"mdn"))
      //获取最后一条数据
      .join(scenicGrid, "grid_id")
      //到达时间
      .withColumn("d_arrive_time", min($"e_time") over Window.partitionBy($"mdn", $"scenic_id"))
      //离开时间
      .withColumn("d_leave_time", max($"l_time") over Window.partitionBy($"mdn", $"scenic_id"))
      //计算停留时间
      .withColumn("d_stay_time", round((unix_timestamp($"d_leave_time", "yyyyMMddHHmmss") - unix_timestamp($"d_arrive_time", "yyyyMMddHHmmss")) / 3600, 2))
      //停留时间大于2小时
      .where($"d_stay_time" > 2)
      .select($"mdn", $"source_city_id", $"departure_time", $"scenic_id", $"d_arrive_time", $"d_leave_time", $"d_stay_time", $"d_max_distance")
      .distinct()

    //常住地在景区内的游客
    val restDF: DataFrame = usertag
      .join(scenicGrid, $"resi_grid_id" === $"grid_id")
      .select($"mdn")

    //去掉常住地在景区内的游客
    val scenicDF: DataFrame = resultDF
      .as("a").join(restDF.as("b"), $"a.mdn" === $"b.mdn", "left")
      .where($"b.mdn".isNull)
      .select($"a.mdn", $"source_city_id", $"departure_time", $"scenic_id", $"d_arrive_time", $"d_leave_time", $"d_stay_time", $"d_max_distance")

    //保存结果
    scenicDF.write
      .format("csv")
      .option("sep", "\u0005")
      .mode(SaveMode.Overwrite)
      .save(s"/daas/subtl/dal/tour/dal_tour_scenic_visitor_msk_d/day_id=$day_id")
  }
}
